package com.jumi.microservice.rocketmq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.jumi.microservice.domain.MessageBody;
import com.jumi.microservice.service.IMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @Description
 * @ClassName MessageMqListener
 * @Author Abel Yang
 * @Date 2020/10/24/024 10:22
 **/

@Component
public class MessageMqListener implements MessageListener {
    private static final Logger log = LoggerFactory.getLogger(MessageMqListener.class);

    @Resource
    private IMessageService iMessageService;

    @Resource
    private RedisTemplate redisTemplate;

    private static final String key = "message::";

    @Override
    public Action consume(Message message, ConsumeContext context) {

        System.out.println("Receive: " + message);
        try {
            Boolean res = redisTemplate.opsForValue().setIfAbsent(key+message.getMsgID(),"消息队列",7, TimeUnit.DAYS);
            if(!res){
                log.info("重复消费：[{}]", message.getMsgID());
                return Action.CommitMessage;
            }
            //do something..
            log.info("MqConsumer[" + message + "]");
            String body = new String(message.getBody(),"utf-8");
            log.info("收到消息[{}]",body);
            MessageBody query = JSONObject.toJavaObject(JSONObject.parseObject(body), MessageBody.class);
//            log.info("MqConsumer body[" + sms.getReceiverMobilePhone()+"]");
            log.info("收到消息转实体[{}]", JSON.toJSONString(query));
            iMessageService.messageSend(query);
            return Action.CommitMessage;
        } catch (Exception e) {
            //删除redis key
            redisTemplate.delete(key+message.getMsgID());
            //消费失败
            log.error("MqConsumer failed[" + e.getMessage() + "]");
            e.printStackTrace();
            return Action.ReconsumeLater;
        }
    }
}

